package cn.slimsmart.thrift.rpc.demo.client;


import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import org.apache.curator.framework.CuratorFramework;
import org.apache.thrift.TException;
import org.apache.thrift.async.AsyncMethodCallback;
import org.apache.thrift.async.TAsyncClientManager;
import org.apache.thrift.protocol.TCompactProtocol;
import org.apache.thrift.protocol.TMultiplexedProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.protocol.TProtocolFactory;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TNonblockingSocket;
import org.apache.thrift.transport.TNonblockingTransport;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;

import cn.slimsmart.thrift.rpc.demo.service.EchoSerivce;
import cn.slimsmart.thrift.rpc.demo.service.HelloWorldService;
import cn.slimsmart.thrift.rpc.demo.service.EchoSerivce.AsyncClient.echo_call;

import com.comtom.soft.zookeeper.ZookeeperFactory;

/**
 * blog http://www.micmiu.com
 *
 * @author Michael
 *
 */
public class MuiltClientZookeepDemo {

	public static final String SERVER_IP = "localhost";
	public static final int SERVER_PORT = 8090;
	public static final int TIMEOUT = 30000;

	/**
	 *
	 * @param userName
	 */
	public void startClient(String userName) {
		CuratorFramework zkClient=ZookeeperFactory.create("127.0.0.1:2181" , 30000, 30000, "rpc/cn.slimsmart.thrift.rpc.demo");
		zkClient.start();
		try {
			zkClient.getData().forPath("/");
			zkClient.getChildren().forPath("/");
			List<String>list=zkClient.getChildren().forPath("/cn.slimsmart.thrift.rpc.demo.service.HelloWorldService");
			list=zkClient.getChildren().forPath("/cn.slimsmart.thrift.rpc.demo.service.HelloWorldService/1.0.0");
			for (String str : list) {
				String seviceClass="cn.slimsmart.thrift.rpc.demo.service.HelloWorldService";
				String[] strArr=str.split(":");
				String ip=strArr[0];
				Integer port=Integer.parseInt(strArr[1]);
				String serviceName=strArr[2];
				startNoBlockClient(ip,port,serviceName,false);
				startHaHsClient(ip,port,serviceName,false);
				startThreadPoolClent(ip,port,serviceName,false);
				startTThreadedSelectorClent(ip,port,serviceName,false);
			}
			list=zkClient.getChildren().forPath("/cn.slimsmart.thrift.rpc.demo.EchoSerivce/1.0.0");
			for (String str : list) {
				String[] strArr=str.split(":");
				String ip=strArr[0];
				Integer port=Integer.parseInt(strArr[1]);
				String serviceName=strArr[2];
				startNoBlockClient(ip,port,serviceName,true);
				startHaHsClient(ip,port,serviceName,true);
				startThreadPoolClent(ip,port,serviceName,true);
				startTThreadedSelectorClent(ip,port,serviceName,true);
			}
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		startClient2("d");
	}

	/**
	 * @param args
	 */
	public static void main(String[] args) {
		MuiltClientZookeepDemo client = new MuiltClientZookeepDemo();
		client.startClient("");

	}

	
	public static void startNoBlockClient(String ip,Integer port,String serviceName,boolean flag){
		//客户端和服务端要使用同一中 Protocol 和 Transport，否则会抛出异常
		try {
			TTransport transport = new TFramedTransport(new TSocket(ip,port, TIMEOUT));
			// 协议要和服务端一致
			TProtocol protocol = new TCompactProtocol(transport);
			transport.open();
			TMultiplexedProtocol mp1 = new TMultiplexedProtocol(protocol,serviceName);
			if(flag){
				EchoSerivce.Client client = new EchoSerivce.Client(mp1);
				String result = client.echo("dddd");
				System.out.println("Thrify client result =: " + result);
			}else{
				HelloWorldService.Client client2 = new HelloWorldService.Client(mp1);
				String result2 = client2.sayHello("hello");
				System.out.println("Thrify client result2 =: " + result2);
			}
			transport.close();  
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	
	public static void startHaHsClient(String ip,Integer port,String serviceName,boolean flag){
		//客户端和服务端要使用同一中 Protocol 和 Transport，否则会抛出异常
		try {
			TTransport transport = new TFramedTransport(new TSocket(ip,port, TIMEOUT));
			// 协议要和服务端一致
			TProtocol protocol = new TCompactProtocol(transport);
			transport.open();
			TMultiplexedProtocol mp1 = new TMultiplexedProtocol(protocol,serviceName);
			if(flag){
				EchoSerivce.Client client = new EchoSerivce.Client(mp1);
				String result = client.echo("dddd");
				System.out.println("Thrify client result =: " + result);
			}else{
				HelloWorldService.Client client2 = new HelloWorldService.Client(mp1);
				String result2 = client2.sayHello("hello");
				System.out.println("Thrify client result2 =: " + result2);
			}
			
			transport.close();  
			
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	
	public static void startThreadPoolClent(String ip,Integer port,String serviceName,boolean flag){
		//客户端和服务端要使用同一中 Protocol 和 Transport，否则会抛出异常
		try {
			TTransport transport = new TFramedTransport(new TSocket(ip,port, TIMEOUT));
			// 协议要和服务端一致
			TProtocol protocol = new TCompactProtocol(transport);
			transport.open();
			TMultiplexedProtocol mp1 = new TMultiplexedProtocol(protocol,serviceName);
			if(flag){
				EchoSerivce.Client client = new EchoSerivce.Client(mp1);
				String result = client.echo("dddd");
				System.out.println("Thrify client result =: " + result);
			}else{
				HelloWorldService.Client client2 = new HelloWorldService.Client(mp1);
				String result2 = client2.sayHello("hello");
				System.out.println("Thrify client result2 =: " + result2);
			}
			transport.close();  
			
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	
	public static void startTThreadedSelectorClent(String ip,Integer port,String serviceName,boolean flag){
		//客户端和服务端要使用同一中 Protocol 和 Transport，否则会抛出异常
		try {
			TTransport transport = new TFramedTransport(new TSocket(ip,port, TIMEOUT));
			// 协议要和服务端一致
			TProtocol protocol = new TCompactProtocol(transport);
			transport.open();
			TMultiplexedProtocol mp1 = new TMultiplexedProtocol(protocol,serviceName);
			if(flag){
				EchoSerivce.Client client = new EchoSerivce.Client(mp1);
				String result = client.echo("dddd");
				System.out.println("Thrify client result =: " + result);
			}else{
				HelloWorldService.Client client2 = new HelloWorldService.Client(mp1);
				String result2 = client2.sayHello("hello");
				System.out.println("Thrify client result2 =: " + result2);
			}
			transport.close();  
			
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	
	
	
	/**
	 *
	 * @param userName
	 */
	public static void startClient2(String userName) {
		try {
			TAsyncClientManager clientManager = new TAsyncClientManager();
			TNonblockingTransport transport = new TNonblockingSocket(SERVER_IP,10004, TIMEOUT);
			TProtocol protocol = new TCompactProtocol(transport);
			TProtocolFactory tprotocol = new TCompactProtocol.Factory();
			
			TMultiplexedProtocol mp1 = new TMultiplexedProtocol(protocol,"echoSerivce");
			TMultiplexedProtocol mp2 = new TMultiplexedProtocol(protocol,"helloWorldService");
			
			EchoSerivce.AsyncClient asyncClient2 = new EchoSerivce.AsyncClient.Factory(clientManager, tprotocol).getAsyncClient(transport);
			
			EchoSerivce.AsyncClient asyncClient = new EchoSerivce.AsyncClient(tprotocol, clientManager, transport);
			
			
			
			System.out.println("Client start .....");

			CountDownLatch latch = new CountDownLatch(1);
			AsynCallback callBack = new AsynCallback(latch);
			System.out.println("call method sayHello start ...");
			asyncClient.echo(userName, callBack);
			System.out.println("call method sayHello .... end");
			boolean wait = latch.await(30, TimeUnit.SECONDS);
			System.out.println("latch.await =:" + wait);
		} catch (Exception e) {
			e.printStackTrace();
		}
		System.out.println("startClient end.");
	}

	public static class AsynCallback implements AsyncMethodCallback<echo_call> {
		private CountDownLatch latch;

		public AsynCallback(CountDownLatch latch) {
			this.latch = latch;
		}

		@Override
		public void onComplete(echo_call response) {
			System.out.println("onComplete");
			try {
				// Thread.sleep(1000L * 1);
				System.out.println("AsynCall result =:"
						+ response.getResult().toString());
			} catch (TException e) {
				e.printStackTrace();
			} catch (Exception e) {
				e.printStackTrace();
			} finally {
				latch.countDown();
			}
		}

		@Override
		public void onError(Exception exception) {
			System.out.println("onError :" + exception.getMessage());
			latch.countDown();
		}

	}
	
}
